package com.intellif.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * @author by mozping
 * @Classname MyKafkaProducer
 * @Description TODO
 * @Date 2019/7/20 9:38
 */
public class MyKafkaProducer {

    public static final String BROKER_LIST = "192.168.13.53:9092";
    public static final String TOPIC = "testTopic-2";
    public static final String CLIENT_ID = "producer-mzp-1";


    public static Properties initConfig() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, CLIENT_ID);
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);
        return properties;
    }

    public static void main(String[] args) {
        sendAndForget();
        //sendSync();
        //sendAync();

    }

    private static void sendAndForget() {
        Properties prop = initConfig();
        KafkaProducer<String, String> producer = new KafkaProducer<>(prop);
        //ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "Hello world , Kafka !" + new Date());
        RecordHeaders headers = new RecordHeaders();
        RecordHeader recordHeader = new RecordHeader("headKey", "headValue".getBytes());
        headers.add(recordHeader);
        ProducerRecord<String, String> record =
                new ProducerRecord<>(TOPIC, 0, System.currentTimeMillis(), null, "Hello world , Kafka !", headers);
        try {
            producer.send(record);
            System.out.println("发送完毕...");
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        producer.flush();
        System.out.println("End ... ");
    }

    private static void sendSync() {
        Properties prop = initConfig();
        KafkaProducer<String, String> producer = new KafkaProducer<>(prop);
        //ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "Hello world , Kafka !" + new Date());
        RecordHeaders headers = new RecordHeaders();
        RecordHeader recordHeader = new RecordHeader("headKey", "headValue".getBytes());
        headers.add(recordHeader);
        ProducerRecord<String, String> record =
                new ProducerRecord<>(TOPIC, 0, System.currentTimeMillis(), null, "Hello world , Kafka !", headers);
        try {
            RecordMetadata recordMetadata = producer.send(record).get();
            System.out.println("发送完毕..." + recordMetadata);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        producer.flush();
        System.out.println("End ... ");
    }

    private static void sendSyncWithFuture() {
        Properties prop = initConfig();
        KafkaProducer<String, String> producer = new KafkaProducer<>(prop);
        //ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "Hello world , Kafka !" + new Date());
        RecordHeaders headers = new RecordHeaders();
        RecordHeader recordHeader = new RecordHeader("headKey", "headValue".getBytes());
        headers.add(recordHeader);
        ProducerRecord<String, String> record =
                new ProducerRecord<>(TOPIC, 0, System.currentTimeMillis(), null, "Hello world , Kafka !", headers);
        try {
            Future<RecordMetadata> future = producer.send(record);
            RecordMetadata recordMetadata = future.get();
            System.out.println("发送完毕..." + recordMetadata);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        producer.flush();
        System.out.println("End ... ");
    }

    private static void sendAync() {
        Properties prop = initConfig();
        KafkaProducer<String, String> producer = new KafkaProducer<>(prop);
        //ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "Hello world , Kafka !" + new Date());
        RecordHeaders headers = new RecordHeaders();
        RecordHeader recordHeader = new RecordHeader("headKey", "headValue".getBytes());
        headers.add(recordHeader);
        ProducerRecord<String, String> record =
                new ProducerRecord<>(TOPIC, 0, System.currentTimeMillis(), null, "Hello world , Kafka !", headers);
        try {
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if(exception!=null){
                        System.out.println("发送消息异常...");
                        exception.printStackTrace();
                    }
                    System.out.println("发送成功后回调，" + metadata);
                }
            });
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        producer.flush();
        System.out.println("End ... ");
    }


}